package com.elitedatai.mchtest.wayline.service.impl;

import com.dji.sdk.cloudapi.device.ExitWaylineWhenRcLostEnum;
import com.dji.sdk.cloudapi.media.UploadFlighttaskMediaPrioritize;
import com.dji.sdk.cloudapi.media.api.AbstractMediaService;
import com.dji.sdk.cloudapi.wayline.*;
import com.dji.sdk.cloudapi.wayline.api.AbstractWaylineService;
import com.dji.sdk.common.HttpResultResponse;
import com.dji.sdk.common.SDKManager;
import com.dji.sdk.mqtt.MqttReply;
import com.dji.sdk.mqtt.events.TopicEventsRequest;
import com.dji.sdk.mqtt.events.TopicEventsResponse;
import com.dji.sdk.mqtt.services.ServicesReplyData;
import com.dji.sdk.mqtt.services.TopicServicesResponse;
import com.elitedatai.mchtest.common.error.CommonErrorEnum;
import com.elitedatai.mchtest.common.model.CustomClaim;
import com.elitedatai.mchtest.component.mqtt.model.EventsReceiver;
import com.elitedatai.mchtest.component.redis.RedisConst;
import com.elitedatai.mchtest.component.redis.RedisOpsUtils;
import com.elitedatai.mchtest.component.websocket.service.IWebSocketMessageService;
import com.elitedatai.mchtest.manage.model.dto.DeviceDTO;
import com.elitedatai.mchtest.manage.service.IDeviceRedisService;
import com.elitedatai.mchtest.media.model.MediaFileCountDTO;
import com.elitedatai.mchtest.media.service.IMediaRedisService;
import com.elitedatai.mchtest.mission.entity.FlightPlan;
import com.elitedatai.mchtest.mission.enums.PlanStatus;
import com.elitedatai.mchtest.mission.service.IFlightPlanService;
import com.elitedatai.mchtest.wayline.model.dto.ConditionalWaylineJobKey;
import com.elitedatai.mchtest.wayline.model.dto.WaylineJobDTO;
import com.elitedatai.mchtest.wayline.model.dto.WaylineTaskConditionDTO;
import com.elitedatai.mchtest.wayline.model.enums.WaylineJobStatusEnum;
import com.elitedatai.mchtest.wayline.model.param.CreateJobParam;
import com.elitedatai.mchtest.wayline.model.param.UpdateJobParam;
import com.elitedatai.mchtest.wayline.service.IFlightTaskService;
import com.elitedatai.mchtest.wayline.service.IWaylineFileService;
import com.elitedatai.mchtest.wayline.service.IWaylineJobService;
import com.elitedatai.mchtest.wayline.service.IWaylineRedisService;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.messaging.MessageHeaders;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import java.net.URL;
import java.sql.SQLException;
import java.time.*;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import com.elitedatai.mchtest.wayline.model.enums.WaylineErrorCodeEnum;



@Service
@Slf4j
public class FlightTaskServiceImpl extends AbstractWaylineService implements IFlightTaskService {

    @Autowired
    private ObjectMapper mapper;

    @Autowired
    private IWebSocketMessageService websocketMessageService;

    @Autowired
    private IWaylineJobService waylineJobService;

    @Autowired
    private IDeviceRedisService deviceRedisService;

    @Autowired
    private IWaylineRedisService waylineRedisService;

    @Autowired
    private IMediaRedisService mediaRedisService;

    @Autowired
    private IWaylineFileService waylineFileService;

    @Autowired
    private IFlightPlanService flightPlanService;

    @Autowired
    private SDKWaylineService abstractWaylineService;

    @Autowired
    @Qualifier("mediaServiceImpl")
    private AbstractMediaService abstractMediaService;

    /**
     * 检查定时任务的调度方法
     * 该方法每5秒执行一次，用于检查Redis中存储的定时航线任务并在合适的时间执行
     */
    @Scheduled(initialDelay = 10, fixedRate = 5, timeUnit = TimeUnit.SECONDS)
    public void checkScheduledJob() {
        // 从Redis有序集合中获取最小分数的任务ID（即最早需要执行的任务）
        Object jobIdValue = RedisOpsUtils.zGetMin(RedisConst.WAYLINE_JOB_TIMED_EXECUTE);
        if (Objects.isNull(jobIdValue)) {
            // 如果没有待执行的定时任务，直接返回
            return;
        }
        log.info("Check the timed tasks of the wayline. {}", jobIdValue);
        
        // 解析任务ID字符串，格式为：{workspace_id}:{dock_sn}:{job_id}
        String[] jobArr = String.valueOf(jobIdValue).split(RedisConst.DELIMITER);
        // 获取该任务的执行时间（时间戳）
        double time = RedisOpsUtils.zScore(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, jobIdValue);
        // 获取当前系统时间
        long now = System.currentTimeMillis();
        // 设置时间偏移量为30秒，用于判断任务是否过期或到达执行时间
        int offset = 30_000;

        // 如果任务执行时间已经过期超过30秒，则直接删除该任务并标记为失败
        if (time < now - offset) {
            // 从Redis中移除过期的定时任务
            RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, jobIdValue);
            // 更新任务状态为失败，设置错误码为请求超时
            waylineJobService.updateJob(WaylineJobDTO.builder()
                    .jobId(jobArr[2])
                    .status(WaylineJobStatusEnum.FAILED.getVal())
                    .executeTime(LocalDateTime.now())
                    .completedTime(LocalDateTime.now())
                    .code(HttpStatus.SC_REQUEST_TIMEOUT).build());
            return;
        }

        // 如果当前时间在任务执行时间的允许范围内（前后30秒），则执行任务
        if (now <= time && time <= now + offset) {
            try {
                // 执行飞行任务，传入工作空间ID和任务ID
                this.executeFlightTask(jobArr[0], jobArr[2]);
            } catch (Exception e) {
                // 如果任务执行失败，记录日志并更新任务状态为失败
                log.info("The scheduled task delivery failed.");
                waylineJobService.updateJob(WaylineJobDTO.builder()
                        .jobId(jobArr[2])
                        .status(WaylineJobStatusEnum.FAILED.getVal())
                        .executeTime(LocalDateTime.now())
                        .completedTime(LocalDateTime.now())
                        .code(HttpStatus.SC_INTERNAL_SERVER_ERROR).build());
            } finally {
                // 无论执行成功或失败，都要从Redis中移除该定时任务
                RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, jobIdValue);
            }
        }
    }

    /**
     * 准备条件任务的调度方法
     * 该方法每5秒执行一次，用于检查条件任务并在合适的时间准备执行
     * 条件任务需要满足特定条件（如电池电量、存储容量等）才能执行
     */
    @Scheduled(initialDelay = 10, fixedRate = 5, timeUnit = TimeUnit.SECONDS)
    public void prepareConditionJob() {
        // 获取最近的条件航线任务键
        Optional<ConditionalWaylineJobKey> jobKeyOpt = waylineRedisService.getNearestConditionalWaylineJob();
        if (jobKeyOpt.isEmpty()) {
            // 如果没有条件任务需要处理，直接返回
            return;
        }
        ConditionalWaylineJobKey jobKey = jobKeyOpt.get();
        log.info("Check the conditional tasks of the wayline. {}", jobKey.toString());
        
        // 获取条件任务的执行时间
        double time = waylineRedisService.getConditionalWaylineJobTime(jobKey);
        // 获取当前系统时间
        long now = System.currentTimeMillis();
        // 提前一天准备任务（86400000毫秒 = 24小时）
        int offset = 86_400_000;

        // 如果当前时间加上偏移量仍小于任务时间，说明还不需要准备，直接返回
        if (now + offset < time) {
            return;
        }

        // 创建一个默认的失败任务对象，用于异常情况下的状态更新
        WaylineJobDTO job = WaylineJobDTO.builder()
                .jobId(jobKey.getJobId())
                .status(WaylineJobStatusEnum.FAILED.getVal())
                .executeTime(LocalDateTime.now())
                .completedTime(LocalDateTime.now())
                .code(HttpStatus.SC_INTERNAL_SERVER_ERROR).build();
        try {
            // 从Redis中获取条件航线任务详情
            Optional<WaylineJobDTO> waylineJobOpt = waylineRedisService.getConditionalWaylineJob(jobKey.getJobId());
            if (waylineJobOpt.isEmpty()) {
                // 如果任务不存在，设置错误码并更新任务状态
                job.setCode(CommonErrorEnum.REDIS_DATA_NOT_FOUND.getCode());
                waylineJobService.updateJob(job);
                // 移除准备中的条件任务
                waylineRedisService.removePrepareConditionalWaylineJob(jobKey);
                return;
            }
            WaylineJobDTO waylineJob = waylineJobOpt.get();

            // 发布单个飞行任务
            HttpResultResponse result = this.publishOneFlightTask(waylineJob);
            // 移除准备中的条件任务
            waylineRedisService.removePrepareConditionalWaylineJob(jobKey);

            // 如果发布成功，直接返回
            if (HttpResultResponse.CODE_SUCCESS == result.getCode()) {
                return;
            }

            // 如果超过结束时间，不再重试
            waylineRedisService.delConditionalWaylineJob(jobKey.getJobId());
            if (waylineJob.getEndTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() - RedisConst.WAYLINE_JOB_BLOCK_TIME * 1000 < now) {
                return;
            }

            // 如果未超过结束时间，则重试准备任务
            this.retryPrepareJob(jobKey, waylineJob);

        } catch (Exception e) {
            // 如果准备条件任务失败，记录日志并更新任务状态
            log.info("Failed to prepare the conditional task.");
            waylineJobService.updateJob(job);
        }

    }

    /**
     * 填充立即执行任务的时间参数
     * 对于立即执行的任务，使用服务器当前时间作为执行时间
     * @param param 创建任务的参数对象
     */
    private void fillImmediateTime(CreateJobParam param) {
        // 如果不是立即执行类型的任务，直接返回，不做任何处理
        if (TaskTypeEnum.IMMEDIATE != param.getTaskType()) {
            return;
        }
        // 获取当前时间戳（秒级）
        long now = System.currentTimeMillis() / 1000;
        // 设置任务执行日期为当前时间
        param.setTaskDays(List.of(now));
        // 设置任务执行时间段为当前时间（开始时间和结束时间都是当前时间）
        param.setTaskPeriods(List.of(List.of(now)));
    }


    /**
     * 为条件任务添加执行条件
     * 条件任务需要满足特定的条件才能执行，如电池电量、存储容量等
     * @param waylineJob 航线任务对象
     * @param param 创建任务的参数对象
     * @param beginTime 任务开始时间
     * @param endTime 任务结束时间
     */
    private void addConditions(WaylineJobDTO waylineJob, CreateJobParam param, Long beginTime, Long endTime) {
        // 如果不是条件任务类型，直接返回，不添加条件
        if (TaskTypeEnum.CONDITIONAL != param.getTaskType()) {
            return;
        }

        // 构建任务条件对象
        waylineJob.setConditions(
                WaylineTaskConditionDTO.builder()
                        // 设置可执行条件：如果指定了最小存储容量，则添加存储容量条件
                        .executableConditions(Objects.nonNull(param.getMinStorageCapacity()) ?
                                new ExecutableConditions().setStorageCapacity(param.getMinStorageCapacity()) : null)
                        // 设置准备条件：包括电池电量、开始时间和结束时间
                        .readyConditions(new ReadyConditions()
                                .setBatteryCapacity(param.getMinBatteryCapacity())
                                .setBeginTime(beginTime)
                                .setEndTime(endTime))
                        .build());

        // 将条件航线任务保存到Redis中
        waylineRedisService.setConditionalWaylineJob(waylineJob);
        // 将任务添加到准备执行的条件任务队列中
        // key格式: wayline_job_condition, value格式: {workspace_id}:{dock_sn}:{job_id}
        boolean isAdd = waylineRedisService.addPrepareConditionalWaylineJob(waylineJob);
        if (!isAdd) {
            // 如果添加失败，抛出运行时异常
            throw new RuntimeException("Failed to create conditional job.");
        }
    }

    /**
     * 发布飞行任务的主要方法
     * 根据任务参数创建并发布飞行任务，支持立即执行、定时执行和条件执行三种类型
     * @param param 创建任务的参数对象，包含任务类型、执行时间、航线文件等信息
     * @param customClaim 用户自定义声明，包含工作空间ID和用户名等信息
     * @return HttpResultResponse 操作结果响应
     * @throws SQLException 数据库操作异常
     */
    @Override
    public HttpResultResponse publishFlightTask(CreateJobParam param, CustomClaim customClaim) throws SQLException {
        // 如果是立即执行任务，填充当前时间作为执行时间
        fillImmediateTime(param);

        // 遍历所有任务执行日期
        for (Long taskDay : param.getTaskDays()) {
            // 将时间戳转换为本地日期
            LocalDate date = LocalDate.ofInstant(Instant.ofEpochSecond(taskDay), ZoneId.systemDefault());
            // 遍历每天的所有执行时间段
            for (List<Long> taskPeriod : param.getTaskPeriods()) {
                // 计算任务开始时间（毫秒级时间戳）
                long beginTime = LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(0)), ZoneId.systemDefault()))
                        .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
                // 计算任务结束时间，如果时间段只有一个时间点，则结束时间等于开始时间
                long endTime = taskPeriod.size() > 1 ?
                        LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(1)), ZoneId.systemDefault()))
                                .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() : beginTime;
                // 对于非立即执行任务，如果结束时间已过期，跳过该时间段
                if (TaskTypeEnum.IMMEDIATE != param.getTaskType() && endTime < System.currentTimeMillis()) {
                    continue;
                }

                // 创建航线任务记录
                Optional<WaylineJobDTO> waylineJobOpt = waylineJobService.createWaylineJob(param, customClaim.getWorkspaceId(), customClaim.getUsername(), beginTime, endTime);

                if (waylineJobOpt.isEmpty()) {
                    throw new SQLException("Failed to create wayline job.");
                }

                WaylineJobDTO waylineJob = waylineJobOpt.get();
                // 如果是条件任务类型，添加任务执行条件
                addConditions(waylineJob, param, beginTime, endTime);

                FlightPlan flightPlan = flightPlanService.getFlightPlanByPlanId(param.getPlanId());
                if (Objects.isNull(flightPlan)) {
                    throw new SQLException("Failed to get flight plan.");
                }
                flightPlan.setJobId(waylineJob.getJobId());
                // 更新飞行计划状态为执行中
                flightPlan.setStatus(PlanStatus.EXECUTING.getVal());
                flightPlanService.save(flightPlan);

                // 发布单个飞行任务
                HttpResultResponse response = this.publishOneFlightTask(waylineJob);
                if (HttpResultResponse.CODE_SUCCESS != response.getCode()) {
                    // 如果发布失败，直接返回错误响应
                    return response;
                }
            }
        }
        // 所有任务都发布成功，返回成功响应
        return HttpResultResponse.success();
    }

    /**
     * 发布单个飞行任务
     * 检查设备在线状态，准备任务，并根据任务类型执行相应的操作
     * @param waylineJob 航线任务数据传输对象，包含任务的所有必要信息
     * @return HttpResultResponse 操作结果响应
     * @throws SQLException 数据库操作异常
     */
    public HttpResultResponse publishOneFlightTask(WaylineJobDTO waylineJob) throws SQLException {

        // 检查机场设备是否在线
        boolean isOnline = deviceRedisService.checkDeviceOnline(waylineJob.getDockSn());
        if (!isOnline) {
            throw new RuntimeException("Dock 离线");
        }

        // 准备飞行任务，包括获取航线文件、构建任务请求等
        boolean isSuccess = this.prepareFlightTask(waylineJob);
        if (!isSuccess) {
            return HttpResultResponse.error("Failed to prepare job.");
        }

        // 如果是立即执行任务，直接执行任务
        if (TaskTypeEnum.IMMEDIATE == waylineJob.getTaskType()) {
            if (!executeFlightTask(waylineJob.getWorkspaceId(), waylineJob.getJobId())) {
                return HttpResultResponse.error("作业执行失败");
            }
        }

        // 如果是定时任务，将任务添加到定时执行队列中
        if (TaskTypeEnum.TIMED == waylineJob.getTaskType()) {
            // Redis键值对格式: key: wayline_job_timed, value: {workspace_id}:{dock_sn}:{job_id}
            // 使用任务开始时间作为分数，用于排序和定时执行
            boolean isAdd = RedisOpsUtils.zAdd(RedisConst.WAYLINE_JOB_TIMED_EXECUTE,
                    waylineJob.getWorkspaceId() + RedisConst.DELIMITER + waylineJob.getDockSn() + RedisConst.DELIMITER + waylineJob.getJobId(),
                    waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
            if (!isAdd) {
                return HttpResultResponse.error("创建定时作业失败");
            }
        }

        // 任务发布成功
        return HttpResultResponse.success();
    }

    /**
     * 准备飞行任务
     * 获取航线文件信息，构建飞行任务准备请求，并发送给设备SDK进行任务准备
     * @param waylineJob 航线任务数据传输对象
     * @return Boolean 准备结果，true表示成功，false表示失败
     * @throws SQLException 数据库操作异常
     */
    private Boolean prepareFlightTask(WaylineJobDTO waylineJob) throws SQLException {
        // 根据航线文件ID获取航线文件信息
        Optional<GetWaylineListResponse> waylineFile = waylineFileService.getWaylineByWaylineId(waylineJob.getWorkspaceId(), waylineJob.getFileId());
        if (waylineFile.isEmpty()) {
            throw new SQLException("航线文件不存在");
        }

        // 获取航线文件的访问URL
        URL url = waylineFileService.getObjectUrl(waylineJob.getWorkspaceId(), waylineFile.get().getId());

        // 构建飞行任务准备请求对象
        FlighttaskPrepareRequest flightTask = new FlighttaskPrepareRequest()
                .setFlightId(waylineJob.getJobId())  // 设置飞行任务ID
                .setExecuteTime(waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli())  // 设置执行时间
                .setTaskType(waylineJob.getTaskType())  // 设置任务类型
                .setWaylineType(waylineJob.getWaylineType())  // 设置航线类型
                .setRthAltitude(waylineJob.getRthAltitude())  // 设置返航高度
                .setOutOfControlAction(waylineJob.getOutOfControlAction())  // 设置失控动作
                .setExitWaylineWhenRcLost(ExitWaylineWhenRcLostEnum.EXECUTE_RC_LOST_ACTION)  // 设置遥控器失联时的动作
                .setFile(new FlighttaskFile()
                        .setUrl(url.toString())  // 设置航线文件URL
                        .setFingerprint(waylineFile.get().getSign()));  // 设置文件指纹用于校验

        // 如果是条件任务，设置执行条件
        if (TaskTypeEnum.CONDITIONAL == waylineJob.getTaskType()) {
            if (Objects.isNull(waylineJob.getConditions())) {
                throw new IllegalArgumentException();
            }
            // 设置准备条件和可执行条件
            flightTask.setReadyConditions(waylineJob.getConditions().getReadyConditions());
            flightTask.setExecutableConditions(waylineJob.getConditions().getExecutableConditions());
        }

        // 通过设备SDK发送任务准备请求
        TopicServicesResponse<ServicesReplyData> serviceReply = abstractWaylineService.flighttaskPrepare(
                SDKManager.getDeviceSDK(waylineJob.getDockSn()), flightTask);
        
        // 检查任务准备结果
        if (!serviceReply.getData().getResult().isSuccess()) {
            log.info("Prepare task ====> Error code: {}", serviceReply.getData().getResult());
            // 任务准备失败，更新任务状态为失败
            waylineJobService.updateJob(WaylineJobDTO.builder()
                    .workspaceId(waylineJob.getWorkspaceId())
                    .jobId(waylineJob.getJobId())
                    .executeTime(LocalDateTime.now())
                    .status(WaylineJobStatusEnum.FAILED.getVal())
                    .completedTime(LocalDateTime.now())
                    .code(serviceReply.getData().getResult().getCode()).build());
            return false;
        }
        // 任务准备成功
        return true;
    }


    /**
     * 执行飞行任务
     * 根据工作空间ID和任务ID执行指定的飞行任务，包括设备在线检查、任务执行请求发送和状态更新
     * @param workspaceId 工作空间ID
     * @param jobId 任务ID
     * @return Boolean 执行结果，true表示成功，false表示失败
     */
    @Override
    public Boolean executeFlightTask(String workspaceId, String jobId) {
        // 根据任务ID获取任务详细信息
        Optional<WaylineJobDTO> waylineJob = waylineJobService.getJobByJobId(workspaceId, jobId);
        if (waylineJob.isEmpty()) {
            throw new IllegalArgumentException("任务不存在");
        }

        // 检查机场设备是否在线
        boolean isOnline = deviceRedisService.checkDeviceOnline(waylineJob.get().getDockSn());
        if (!isOnline) {
            throw new RuntimeException("Dock 离线");
        }

        WaylineJobDTO job = waylineJob.get();

        // 通过设备SDK发送任务执行请求
        TopicServicesResponse<ServicesReplyData> serviceReply = abstractWaylineService.flighttaskExecute(
                SDKManager.getDeviceSDK(job.getDockSn()), new FlighttaskExecuteRequest().setFlightId(jobId));
        
        // 检查任务执行请求结果
        if (!serviceReply.getData().getResult().isSuccess()) {
            log.info("Execute job ====> Error: {}", serviceReply.getData().getResult());
            // 任务执行失败，更新任务状态为失败
            waylineJobService.updateJob(WaylineJobDTO.builder()
                    .jobId(jobId)
                    .executeTime(LocalDateTime.now())
                    .status(WaylineJobStatusEnum.FAILED.getVal())
                    .completedTime(LocalDateTime.now())
                    .code(serviceReply.getData().getResult().getCode()).build());
            
            // 如果是条件任务执行失败且错误码表示阻塞，则将任务设置为阻塞状态
            if (TaskTypeEnum.CONDITIONAL == job.getTaskType()
                    && WaylineErrorCodeEnum.find(serviceReply.getData().getResult().getCode()).isBlock()) {
                waylineRedisService.setBlockedWaylineJob(job.getDockSn(), jobId);
            }
            return false;
        }

        // 任务执行成功，更新任务状态为进行中
        waylineJobService.updateJob(WaylineJobDTO.builder()
                .jobId(jobId)
                .executeTime(LocalDateTime.now())
                .status(WaylineJobStatusEnum.IN_PROGRESS.getVal())
                .build());
        
        // 在Redis中设置正在运行的任务信息，用于任务进度跟踪
        waylineRedisService.setRunningWaylineJob(job.getDockSn(), EventsReceiver.<FlighttaskProgress>builder().bid(jobId).sn(job.getDockSn()).build());
        return true;
    }

    /**
     * 取消飞行任务
     * 批量取消指定工作空间下的多个飞行任务，只能取消状态为PENDING的任务
     * @param workspaceId 工作空间ID
     * @param jobIds 要取消的任务ID集合
     */
    @Override
    public void cancelFlightTask(String workspaceId, Collection<String> jobIds) {
        // 获取指定工作空间下状态为PENDING的任务列表
        List<WaylineJobDTO> waylineJobs = waylineJobService.getJobsByConditions(workspaceId, jobIds, WaylineJobStatusEnum.PENDING);

        // 提取可取消任务的ID集合
        Set<String> waylineJobIds = waylineJobs.stream().map(WaylineJobDTO::getJobId).collect(Collectors.toSet());
        
        // 检查任务状态是否正确，只有PENDING状态的任务才能被取消
        boolean isErr = !jobIds.removeAll(waylineJobIds) || !jobIds.isEmpty() ;
        if (isErr) {
            throw new IllegalArgumentException("这个任务状态不正确，只能取消PENDING状态的任务。" + Arrays.toString(jobIds.toArray()));
        }

        // 按机场设备序列号对任务进行分组，同一机场的任务可以批量取消
        Map<String, List<String>> dockJobs = waylineJobs.stream()
                .collect(Collectors.groupingBy(WaylineJobDTO::getDockSn,
                        Collectors.mapping(WaylineJobDTO::getJobId, Collectors.toList())));
        
        // 对每个机场的任务分别发送取消请求
        dockJobs.forEach((dockSn, idList) -> this.publishCancelTask(workspaceId, dockSn, idList));

    }

    /**
     * 发布取消任务请求
     * 向指定机场设备发送取消任务请求，并更新任务状态
     * @param workspaceId 工作空间ID
     * @param dockSn 机场设备序列号
     * @param jobIds 要取消的任务ID列表
     */
    public void publishCancelTask(String workspaceId, String dockSn, List<String> jobIds) {
        // 检查机场设备是否在线
        boolean isOnline = deviceRedisService.checkDeviceOnline(dockSn);
        if (!isOnline) {
            throw new RuntimeException("Dock is offline.");
        }

        // 通过设备SDK发送任务取消请求
        TopicServicesResponse<ServicesReplyData> serviceReply = abstractWaylineService.flighttaskUndo(SDKManager.getDeviceSDK(dockSn),
                new FlighttaskUndoRequest().setFlightIds(jobIds));
        if (!serviceReply.getData().getResult().isSuccess()) {
            log.info("Cancel job ====> Error: {}", serviceReply.getData().getResult());
            throw new RuntimeException("Failed to cancel the wayline job of " + dockSn);
        }

        // 更新每个任务的状态为已取消，并从定时任务队列中移除
        for (String jobId : jobIds) {
            waylineJobService.updateJob(WaylineJobDTO.builder()
                    .workspaceId(workspaceId)
                    .jobId(jobId)
                    .status(WaylineJobStatusEnum.CANCEL.getVal())
                    .completedTime(LocalDateTime.now())
                    .build());
            // 从Redis定时任务队列中移除已取消的任务
            RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, workspaceId + RedisConst.DELIMITER + dockSn + RedisConst.DELIMITER + jobId);
        }

    }

    /**
     * 设置媒体文件上传最高优先级
     * 为指定任务的媒体文件设置最高上传优先级，确保该任务的媒体文件优先上传
     * @param workspaceId 工作空间ID
     * @param jobId 任务ID
     */
    @Override
    public void uploadMediaHighestPriority(String workspaceId, String jobId) {
        // 根据任务ID获取任务信息
        Optional<WaylineJobDTO> jobOpt = waylineJobService.getJobByJobId(workspaceId, jobId);
        if (jobOpt.isEmpty()) {
            throw new RuntimeException(CommonErrorEnum.ILLEGAL_ARGUMENT.getMessage());
        }

        String dockSn = jobOpt.get().getDockSn();
        String key = RedisConst.MEDIA_HIGHEST_PRIORITY_PREFIX + dockSn;
        
        // 检查Redis中是否已存在相同任务的优先级设置，如果存在则直接返回
        if (RedisOpsUtils.checkExist(key) && jobId.equals(((MediaFileCountDTO) RedisOpsUtils.get(key)).getJobId())) {
            return;
        }

        // 通过设备SDK发送媒体文件优先级设置请求
        TopicServicesResponse<ServicesReplyData> reply = abstractMediaService.uploadFlighttaskMediaPrioritize(
                SDKManager.getDeviceSDK(dockSn), new UploadFlighttaskMediaPrioritize().setFlightId(jobId));
        if (!reply.getData().getResult().isSuccess()) {
            throw new RuntimeException("Failed to set media job upload priority. Error: " + reply.getData().getResult());
        }
    }

    /**
     * 更新任务状态
     * 根据参数更新任务状态，支持暂停和恢复操作
     * @param workspaceId 工作空间ID
     * @param jobId 任务ID
     * @param param 更新任务参数，包含要更新的状态
     */
    @Override
    public void updateJobStatus(String workspaceId, String jobId, UpdateJobParam param) {
        // 根据任务ID获取任务信息
        Optional<WaylineJobDTO> waylineJobOpt = waylineJobService.getJobByJobId(workspaceId, jobId);
        if (waylineJobOpt.isEmpty()) {
            throw new RuntimeException("任务不存在");
        }
        
        WaylineJobDTO waylineJob = waylineJobOpt.get();
        // 获取当前任务的状态
        WaylineJobStatusEnum statusEnum = waylineJobService.getWaylineState(waylineJob.getDockSn());
        
        // 检查任务状态是否允许操作（已结束或待执行状态不允许暂停/恢复操作）
        if (statusEnum.getEnd() || WaylineJobStatusEnum.PENDING == statusEnum) {
            throw new RuntimeException("The wayline job status does not match, and the operation cannot be performed.");
        }

        // 根据参数中的状态执行相应操作
        switch (param.getStatus()) {
            case PAUSE:
                // 暂停任务
                pauseJob(workspaceId, waylineJob.getDockSn(), jobId, statusEnum);
                break;
            case RESUME:
                // 恢复任务
                resumeJob(workspaceId, waylineJob.getDockSn(), jobId, statusEnum);
                break;
        }

    }

    /**
     * 暂停任务
     * 向设备发送暂停任务请求，并更新Redis中的任务状态
     * @param workspaceId 工作空间ID
     * @param dockSn 机场设备序列号
     * @param jobId 任务ID
     * @param statusEnum 当前任务状态
     */
    private void pauseJob(String workspaceId, String dockSn, String jobId, WaylineJobStatusEnum statusEnum) {
        // 如果任务已经是暂停状态且是当前任务，直接设置暂停状态并返回
        if (WaylineJobStatusEnum.PAUSED == statusEnum && jobId.equals(waylineRedisService.getPausedWaylineJobId(dockSn))) {
            waylineRedisService.setPausedWaylineJob(dockSn, jobId);
            return;
        }

        // 通过设备SDK发送暂停任务请求
        TopicServicesResponse<ServicesReplyData> reply = abstractWaylineService.flighttaskPause(SDKManager.getDeviceSDK(dockSn));
        if (!reply.getData().getResult().isSuccess()) {
            throw new RuntimeException("Failed to pause wayline job. Error: " + reply.getData().getResult());
        }
        
        // 从Redis中删除正在运行的任务记录，并设置为暂停状态
        waylineRedisService.delRunningWaylineJob(dockSn);
        waylineRedisService.setPausedWaylineJob(dockSn, jobId);
    }

    /**
     * 恢复任务
     * 向设备发送恢复任务请求，并更新Redis中的任务状态
     * @param workspaceId 工作空间ID
     * @param dockSn 机场设备序列号
     * @param jobId 任务ID
     * @param statusEnum 当前任务状态
     */
    private void resumeJob(String workspaceId, String dockSn, String jobId, WaylineJobStatusEnum statusEnum) {
        // 获取正在运行的任务信息
        Optional<EventsReceiver<FlighttaskProgress>> runningDataOpt = waylineRedisService.getRunningWaylineJob(dockSn);
        
        // 如果任务已经是进行中状态且是当前任务，直接设置运行状态并返回
        if (WaylineJobStatusEnum.IN_PROGRESS == statusEnum && jobId.equals(runningDataOpt.map(EventsReceiver::getSn).get())) {
            waylineRedisService.setRunningWaylineJob(dockSn, runningDataOpt.get());
            return;
        }
        
        // 通过设备SDK发送恢复任务请求
        TopicServicesResponse<ServicesReplyData> reply = abstractWaylineService.flighttaskRecovery(SDKManager.getDeviceSDK(dockSn));
        if (!reply.getData().getResult().isSuccess()) {
            throw new RuntimeException("Failed to resume wayline job. Error: " + reply.getData().getResult());
        }

        // 如果有正在运行的任务数据，恢复运行状态并删除暂停状态
        runningDataOpt.ifPresent(runningData -> waylineRedisService.setRunningWaylineJob(dockSn, runningData));
        waylineRedisService.delPausedWaylineJob(dockSn);
    }

    /**
     * 重试准备任务
     * 为条件任务创建子任务并重新加入准备队列，用于处理条件任务失败后的重试逻辑
     * @param jobKey 条件任务键，包含工作空间ID和任务ID
     * @param waylineJob 原始航线任务对象
     */
    @Override
    public void retryPrepareJob(ConditionalWaylineJobKey jobKey, WaylineJobDTO waylineJob) {
        // 基于父任务创建子任务
        Optional<WaylineJobDTO> childJobOpt = waylineJobService.createWaylineJobByParent(jobKey.getWorkspaceId(), jobKey.getJobId());
        if (childJobOpt.isEmpty()) {
            log.error("Failed to create wayline job.");
            return;
        }

        WaylineJobDTO newJob = childJobOpt.get();
        // 设置新任务的开始时间为当前时间加上阻塞时间间隔
        newJob.setBeginTime(LocalDateTime.now().plusSeconds(RedisConst.WAYLINE_JOB_BLOCK_TIME));
        
        // 将新任务添加到条件任务准备队列
        boolean isAdd = waylineRedisService.addPrepareConditionalWaylineJob(newJob);
        if (!isAdd) {
            log.error("Failed to create wayline job. {}", newJob.getJobId());
            return;
        }

        // 更新原任务的ID为新任务ID，并重新设置条件任务
        waylineJob.setJobId(newJob.getJobId());
        waylineRedisService.setConditionalWaylineJob(waylineJob);
    }


    /**
     * 处理飞行任务就绪事件
     * 当设备报告任务就绪时，检查并执行被阻塞的条件任务
     * @param response 包含就绪任务ID列表的事件请求
     * @param headers 消息头信息
     * @return MQTT回复响应
     */
    @Override
    public TopicEventsResponse<MqttReply> flighttaskReady(TopicEventsRequest<FlighttaskReady> response, MessageHeaders headers) {
        List<String> flightIds = response.getData().getFlightIds();

        log.info("ready task list：{}", Arrays.toString(flightIds.toArray()) );
        
        // 检查条件任务阻塞状态
        String blockedId = waylineRedisService.getBlockedWaylineJobId(response.getGateway());
        if (!StringUtils.hasText(blockedId)) {
            return null;
        }

        // 获取设备在线信息
        Optional<DeviceDTO> deviceOpt = deviceRedisService.getDeviceOnline(response.getGateway());
        if (deviceOpt.isEmpty()) {
            return null;
        }
        DeviceDTO device = deviceOpt.get();

        try {
            // 遍历就绪的任务ID列表
            for (String jobId : flightIds) {
                // 尝试执行飞行任务
                boolean isExecute = this.executeFlightTask(device.getWorkspaceId(), jobId);
                if (!isExecute) {
                    return null;
                }
                
                // 获取条件任务信息
                Optional<WaylineJobDTO> waylineJobOpt = waylineRedisService.getConditionalWaylineJob(jobId);
                if (waylineJobOpt.isEmpty()) {
                    log.info("The conditional job has expired and will no longer be executed.");
                    return new TopicEventsResponse<>();
                }
                
                // 重试准备条件任务
                WaylineJobDTO waylineJob = waylineJobOpt.get();
                this.retryPrepareJob(new ConditionalWaylineJobKey(device.getWorkspaceId(), response.getGateway(), jobId), waylineJob);
                return new TopicEventsResponse<>();
            }
        } catch (Exception e) {
            log.error("Failed to execute conditional task.");
            e.printStackTrace();
        }
        return new TopicEventsResponse<>();
    }

}